#include <functional>     // std::bind()
#include <unistd.h>       // write()
#include <sys/sendfile.h> // sendfile()
// #include <string>
// #include <errno.h>
// #include <sys/types.h>
// #include <sys/socket.h>
// #include <string.h> <strings.h>
// #include <netinet/tcp.h>
// #include <fcntl.h>  // for open

#include "./TcpConnection.h"
#include "./EventLoop.h"
#include "../utility/Logger.h"
#include "../net/Socket.h"
#include "./Channel.h"

// TODO: 还是会继续loop为空逻辑，优化loop为空的处理
// TODO: TcpServer也有一个相同函数不同输出，不用静态多个会有名字冲突，但可以写成成员/全局方法
static EventLoop *CheckLoopNotNull(EventLoop *loop)
{
    if (!loop)
    {
        LOG_FATAL("%s:%s:%d TcpConnection Loop is null! \n", __FILE__, __FUNCTION__, __LINE__);
    }
    return loop;
}

// OPTIMIZE: new -> make_unique<>
TcpConnection::TcpConnection(EventLoop *loop,
                             const std::string &name,
                             int sockFd,
                             const InetAddress &localAddr,
                             const InetAddress &peerAddr)
    : loop_(CheckLoopNotNull(loop)),
      name_(name),
      socket_(std::make_unique<Socket>(sockFd)),
      channel_(std::make_unique<Channel>(loop, sockFd)),
      localAddr_(localAddr),
      peerAddr_(peerAddr),
      state_(static_cast<int>(ConnState::kConnecting)),
      isRead_(true)
{
    // 给channel注册回调
    this->channel_->setReadCallback(std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
    this->channel_->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
    this->channel_->setCloseCallback(std::bind(&TcpConnection::handleClose, this));
    this->channel_->setErrorCallback(std::bind(&TcpConnection::handleError, this));

    LOG_INFO("TcpConnection::ctor[%s] at fd=%d\n", this->name_.c_str(), sockFd);

    this->socket_->setKeepAlive(true);
}

TcpConnection::~TcpConnection()
{
    // OPTIMIZE: (int) -> static_cast<int>
    // TODO: 对于连接状态，日志输出字符串而不是数字信息更明确
    LOG_INFO("TcpConnection::dtor[%s] at fd=%d state=%d\n", this->name_.c_str(), this->channel_->getFd(), static_cast<int>(this->state_));
}

EventLoop *TcpConnection::getLoop() const
{
    return this->loop_;
}

const std::string &TcpConnection::getName() const
{
    return this->name_;
}

const InetAddress &TcpConnection::getLocalAddr() const
{
    return this->localAddr_;
}

const InetAddress &TcpConnection::getPeerAddr() const
{
    return this->peerAddr_;
}

bool TcpConnection::isConnected() const
{
    return this->state_ == static_cast<int>(ConnState::kConnected);
}

void TcpConnection::setConnectionCallback(const ConnectionCallback &cb)
{
    this->connectionCallback_ = cb;
}

void TcpConnection::setReadWriteCallback(const ReadWriteCallback &cb)
{
    this->readWriteCallback_ = cb;
}

void TcpConnection::setWriteCompleteCallback(const WriteCompleteCallback &cb)
{
    this->writeCompleteCallback_ = cb;
}

void TcpConnection::setCloseCallback(const CloseCallback &cb)
{
    this->closeCallback_ = cb;
}

void TcpConnection::setHighWaterMarkCallback(const HighWaterMarkCallback &cb, size_t highWaterMark)
{
    this->highWaterMarkCallback_ = cb;
    this->highWaterMark_ = highWaterMark;
}

void TcpConnection::setState(ConnState state)
{
    this->state_ = static_cast<int>(state);
}

void TcpConnection::connectEstablished()
{
    // 过程三件套
    // 1. 状态处理
    // 2. channel处理
    // 3. 回调处理
    this->setState(ConnState::kConnected);

    // 可以生成std::shared_ptr<TcpConnection>，给channel的weak_ptr<>持有延长生命周期而不会意外析构
    // channel的回调是tcpconnection注册的，需要保证tcpconnection存在
    this->channel_->tieChannel(shared_from_this());
    this->channel_->enableReadEvent(); // eventloop->poller注册channel读事件

    // 新连接建立，执行用户/开发者回调
    // TODO: 没判断回调是否有效就直接调用了
    this->connectionCallback_(shared_from_this());
}

void TcpConnection::connectDestroyed()
{
    if (this->state_ == static_cast<int>(ConnState::kConnected))
    {
        this->setState(ConnState::kDisconnected);

        this->channel_->disableAllEvent(); // eventloop->poller删除感兴趣事件

        // TODO: 约定判断状态，连接建立/关闭回调是同一个，可以优化分开，换成closeCallback_等
        this->connectionCallback_(shared_from_this());
    }

    this->channel_->removeChannel(); // eventloop->poller删除channel
}

void TcpConnection::connectShutdown()
{
    if (this->state_ == static_cast<int>(ConnState::kConnected))
    {
        this->setState(ConnState::kDisconnecting);

        // 线程安全：操作TcpConnection本身，需要将回调传出去，在对应线程里执行
        this->loop_->runCallback(std::bind(&TcpConnection::connectShutdownInLoop, this));
    }
}

void TcpConnection::connectShutdownInLoop()
{
    if (!this->channel_->isWriteEvent()) // 没有注册写事件，说明outputBuffer_数据向外发送完
    {
        this->socket_->shutdownWrite(); // 关闭写端
    }
}

void TcpConnection::handleRead(Timestamp recvTime)
{
    int savedErrno{0};

    ssize_t n = this->inputBuffer_.readFd(this->channel_->getFd(), &savedErrno);
    if (n > 0) // 有数据
    {
        // 用户/开发者注册的回调
        this->readWriteCallback_(shared_from_this(), &this->inputBuffer_, recvTime);
    }
    else if (n == 0) // 连接断开，一般是客户端
    {
        this->handleClose();
    }
    else // 错误
    {
        errno = savedErrno;

        LOG_ERROR("TcpConnection::handleRead");

        this->handleError();
    }
}

void TcpConnection::handleWrite()
{
    if (this->channel_->isWriteEvent())
    {
        int savedErrno{0};

        // muduo没有封装write()，这里封装了
        // 库本身/服务端->buffer->内核，write()时从有数据可读区读
        ssize_t n = this->outputBuffer_.writeFd(this->channel_->getFd(), &savedErrno);
        if (n > 0)
        {
            this->outputBuffer_.readableIndexReset(n); // 从有数据可读区读了，修改索引

            if (this->outputBuffer_.getReadableSize() == 0) // 没有可读->向fd写完
            {
                this->channel_->disableWriteEvent();

                // 用户/开发者注册的回调
                if (this->writeCompleteCallback_)
                {
                    // 施磊老师认为既然进了这个函数，就是它所属的线程，不应该queueInLoop()
                    // 我认为：前面处理读回调也不是这样，应该统一
                    // 评论认为这个不是要紧事，就先加入队列慢慢执行
                    this->loop_->queueInLoop(std::bind(this->writeCompleteCallback_, shared_from_this()));
                }

                // 若之前connectShutDown()过，会设置kDisconnecting，如果没有写完不会写关闭
                // 这里写完再判断继续写关闭
                if (this->state_ == static_cast<int>(ConnState::kDisconnecting))
                {
                    this->connectShutdownInLoop();
                }
            }
        }
        else // == 0是写0啥都没写
        {
            LOG_ERROR("TcpConnection::handleWrite");
        }
    }
    else
    {
        LOG_ERROR("TcpConnection fd=%d is down, no more writing", this->channel_->getFd());
    }
}

void TcpConnection::handleClose()
{
    LOG_INFO("TcpConnection::handleClose fd=%d state=%d\n", this->channel_->getFd(), static_cast<int>(this->state_));

    this->setState(ConnState::kDisconnected);

    this->channel_->disableAllEvent();

    TcpConnectionPtr connPtr(shared_from_this());
    // 约定判断状态，连接建立/关闭回调是同一个，可以优化分开，换成closeCallback_
    // 没判断回调是否有效就直接调用了
    this->connectionCallback_(connPtr); // 用户/开发者回调
    this->closeCallback_(connPtr);      // TcpServer::removeConnection()
}

void TcpConnection::handleError()
{
    int optVal{};
    socklen_t optLen = sizeof(optVal);
    int err{0};
    if (::getsockopt(this->channel_->getFd(), SOL_SOCKET, SO_ERROR, &optVal, &optLen) < 0)
    {
        err = errno;
    }
    else
    {
        err = optVal;
    }

    LOG_ERROR("TcpConnection::handleError name:%s - SO_ERROR:%d\n", this->name_.c_str(), err);
}

void TcpConnection::sendData(const std::string &buf)
{
    if (this->state_ == static_cast<int>(ConnState::kConnected))
    {
        // 实际上都是执行TcpConnection::sendDataInLoop()
        // 1. 单线程时，用户/开发者调用conn->sendData()，在mainLoop_、mainThread直接执行
        // 2. 多线程时，线程安全：操作TcpConnection本身，需要将回调传出去，在对应线程里执行
        if (this->loop_->isLoopInThread())
        {
            // 已经发buf过去不用管修改索引了
            // 施磊老师说这里也可以queueinloop()/runInLoop
            // 意思就是直接runCallback()，不用if-else判断，runCallback()会判断是否在所属线程里执行
            this->sendDataInLoop(buf.c_str(), buf.size());
        }
        else
        {
            this->loop_->runCallback(std::bind(&TcpConnection::sendDataInLoop, this, buf.c_str(), buf.size()));
        }
    }
}

void TcpConnection::sendDataInLoop(const void *data, size_t needWriteSize)
{
    // TODO：可以判断是写关闭，也不能再发送数据了
    if (this->state_ == static_cast<int>(ConnState::kDisconnected))
    {
        LOG_ERROR("disconnected, give up writing");
        return;
    }

    ssize_t wroteSize{0};
    size_t remainWriteSize{needWriteSize};
    bool hasErr{false};
    // channel_第一次开始写数据，缓冲区没有待读取后write()的数据（库本身/服务端->buffer->内核，write()时从有数据可读区读）
    // 则直接write()出
    if (!this->channel_->isWriteEvent() && this->outputBuffer_.getReadableSize() == 0)
    {
        wroteSize = ::write(this->channel_->getFd(), data, needWriteSize);
        if (wroteSize >= 0)
        {
            remainWriteSize = needWriteSize - wroteSize;
            if (remainWriteSize == 0 && this->writeCompleteCallback_)
            {
                // 数据全部发送完成，可以不用再给channel->poller设置epollout事件
                // × 这里是调用用户/开发者回调
                // 应该也可用runCallback()自己判断执行回调
                // × 这个方法已经交给runCallback()判断执行了，可以直接执行或不着急就直接加入队列
                this->loop_->queueInLoop(std::bind(this->writeCompleteCallback_, shared_from_this()));
            }
        }
        else // writeSize < 0
        {
            wroteSize = 0;
            if (errno != EWOULDBLOCK) // EAGAIN不是非阻塞的正常错误/真正的错误
            {
                LOG_ERROR("TcpConnection::sendInLoop");

                if (errno == EPIPE || errno == ECONNRESET) // SIGPIPE RESET
                {
                    hasErr = true;
                }
            }
        }
    }

    // 说明当前这一次write，没有把数据全部发送，剩余的数据需要保存到缓冲区
    // 然后给channel注册epollout事件
    // poller发现发送缓冲区有空间，会通知相应的sock-channel，调用writeCallback_回调
    // 即调用TcpConnection::handleWrite（），把发送缓冲区中的数据全部发送完成
    if (!hasErr && remainWriteSize > 0)
    {
        // 水位检查
        size_t needWriteSize2 = this->outputBuffer_.getReadableSize();
        // 约定高水位条件
        // 缓冲区有数据可读区/待发送的+当前未发送完的 >=
        // needWriteSize2 < this->highWaterMark_：还要判断这个，如果>实际上次已经超过水位执行过一次水位线回调了
        if (needWriteSize2 + remainWriteSize >= this->highWaterMark_ &&
            needWriteSize2 < this->highWaterMark_ &&
            this->highWaterMarkCallback_)
        {
            this->loop_->queueInLoop(std::bind(this->highWaterMarkCallback_, shared_from_this(), needWriteSize2 + remainWriteSize));
        }

        // 未写完的加入缓冲区
        // 虽然是要加入有数据可读区，但可以用加入空闲可写区的扩容逻辑处理，写不完时重新加入空闲可写区
        // OPTIMIZE: (char *) -> static_cast<const char *>
        this->outputBuffer_.appendWrite(static_cast<const char *>(data) + wroteSize, remainWriteSize);
        if (!this->channel_->isWriteEvent())
        {
            this->channel_->enableWriteEvent(); // 注意注册
        }
    }
}

void TcpConnection::sendDataZeroCopy(int fd, off_t offset, size_t count)
{
    // 类似sendData()
    // this->state_ == static_cast<int>(ConnState::kConnected)
    if (this->isConnected())
    {
        if (this->loop_->isLoopInThread())
        {
            this->sendDataZeroCopyInLoop(fd, offset, count);
        }
        else
        {
            this->loop_->runCallback(std::bind(&TcpConnection::sendDataZeroCopyInLoop, shared_from_this(), fd, offset, count));
        }
    }
    else
    {
        LOG_ERROR("TcpConnection::sendFile - not connected");
    }
}

void TcpConnection::sendDataZeroCopyInLoop(int fd, off_t offset, size_t needWriteSize)
{
    // if (this->state_ == static_cast<int>(ConnState::kDisconnecting))
    if (this->state_ == static_cast<int>(ConnState::kDisconnected))
    {
        LOG_ERROR("disconnected, give up writing");
        return;
    }

    ssize_t wroteSize{0};
    size_t remainWriteSize{needWriteSize};
    bool hasErr{false};
    if (!this->channel_->isWriteEvent() && this->outputBuffer_.getReadableSize() == 0)
    {
        wroteSize = sendfile(this->socket_->getFd(), fd, &offset, remainWriteSize);
        if (wroteSize >= 0)
        {
            remainWriteSize -= wroteSize;
            if (remainWriteSize == 0 && this->writeCompleteCallback_)
            {
                this->loop_->queueInLoop(std::bind(this->writeCompleteCallback_, shared_from_this()));
            }
        }
        else // wroteSize < 0
        {
            wroteSize = 0;
            if (errno != EWOULDBLOCK)
            {
                LOG_ERROR("TcpConnection::sendFileInLoop");

                if (errno == EPIPE || errno == ECONNRESET)
                {
                    hasErr = true;
                }
            }
        }
    }

    // 处理剩余数据
    if (!hasErr && remainWriteSize > 0)
    {
        // 继续发送剩余数据，零拷贝不需要再加入用户缓冲区！
        loop_->queueInLoop(std::bind(&TcpConnection::sendDataZeroCopyInLoop, shared_from_this(), fd, offset, remainWriteSize));
    }
}